package com.second.taobao.topic;

import com.second.util.HbaseUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.BufferedMutator;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.Iterator;

public class SaveUserTopic {


    public static void main(String[] args) {
        SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
        String time = df.format(new Date()) + "000000";
        if (args.length > 0) {
            time = args[0] + "000000";
        }

        SparkSession spark = SparkSession.builder()
                            .enableHiveSupport()
                            .getOrCreate();

//        spark.sql("show databases").show();
        spark.sql("use testdb");
//        spark.sql("show tables").show();
//        spark.sql("select * from behavior_log").show();

        Long now = 0L;
        Long ago_1 = 0L;
        Long ago_7 = 0L;
        Long ago_14 = 0L;
        try {
            df = new SimpleDateFormat("yyyyMMddHHmmss");
            Date date = df.parse(time);

            // 开始处理时间
            Calendar cal = Calendar.getInstance();
            cal.setTime(date);
            now = cal.getTimeInMillis();

            // 过去1天
            cal.setTime(date);
            cal.add(Calendar.DAY_OF_MONTH, -1);
            ago_1 = cal.getTimeInMillis() / 1000;

            // 过去7天
            cal.setTime(date);
            cal.add(Calendar.DAY_OF_MONTH, -7);
            ago_7 = cal.getTimeInMillis() / 1000;

            // 过去14天
            cal.setTime(date);
            cal.add(Calendar.DAY_OF_MONTH, -14);
            ago_14 = cal.getTimeInMillis() / 1000;
        } catch (ParseException e) {
            e.printStackTrace();
        }

        // 用户标签
        Dataset<Row> uinfo = spark.sql("select l.userid as userid, sum(IF(l.time_stamp >= " + ago_7 +
                                 " and l.btag = 'pv', 1, 0)) as last_7_day_review, " +
                                 "sum(IF(l.btag = 'pv', 1, 0)) as last_14_day_review, " +
                                 "sum(IF(l.time_stamp >= " + ago_7 +
                                 " and l.btag = 'buy', 1, 0)) as last_7_day_buy, " +
                                 "sum(IF(l.btag = 'buy', 1, 0)) as last_14_day_buy, " +
                                 "u.age_level as age_level, " +
                                 "u.pvalue_level as pvalue_level, " +
                                 "u.shopping_level as shopping_level, " +
                                 "u.occupation as occupation, " +
                                 "u.new_user_class_level as new_user_class_level " +
                          "from user_profile as u, behavior_log l " +
                          "where u.userid = l.userid " +
                            "and l.time_stamp between " + ago_14 + " and " + now +
                            " and l.btag in ('pv', 'buy')" +
                          " group by l.userid, u.age_level, u.pvalue_level, u.shopping_level, " +
                                    "u.occupation, u.new_user_class_level");
//        uinfo.show(10);

        // 广告标签
        Dataset<Row> adinfo = spark.sql("select a.adgroup_id as adgroup_id, " +
                                 "sum(IF(r.time_stamp >= " + ago_1 + ", 1, 0)) as last_1_day_display, " +
                                 "count(1) as last_7_days_display, " +
                                 "sum(IF(r.time_stamp > " + ago_1 + " and r.clk = '1', 1, 0)) as last_1_day_clicked, " +
                                 "sum(IF(r.clk = '1', 1, 0)) as last_7_day_clicked " +
                            "from raw_sample r, ad_feature a " +
                           "where r.adgroup_id = a.adgroup_id " +
                             "and r.time_stamp between " + ago_7 + " and " + now +
                           " group by a.adgroup_id");
//        adinfo.show(10);

        // HBase存储
        JavaRDD<Row> urdd = uinfo.toJavaRDD();
        urdd.foreachPartition(new VoidFunction<Iterator<Row>>() {
            @Override
            public void call(Iterator<Row> it) {
                Connection connection = null;
                BufferedMutator bufferedMutator = null;
                try {
                    connection = ConnectionFactory.createConnection(HbaseUtils.getHBaseConfig());
                    bufferedMutator = connection.getBufferedMutator(TableName.valueOf(HbaseUtils.TABLE_NAME_USER));

                    while (it.hasNext()){
                        Row row = it.next();
                        Put put = new Put(Bytes.toBytes(row.get(0).toString()));

                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_USER),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_USER_AGE), Bytes.toBytes(row.get(5).toString()));
                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_USER),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_USER_OCCUPATION), Bytes.toBytes(row.get(8).toString()));
                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_USER),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_USER_CITY), Bytes.toBytes(row.get(9).toString()));

                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_SALE),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_SLAE_PVALUE), Bytes.toBytes(row.get(6).toString()));
                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_SALE),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_SLAE_SHOP), Bytes.toBytes(row.get(7).toString()));

                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_BEHAVIOR),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_BEHAVIOR_RV_7), Bytes.toBytes(row.get(1).toString()));
                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_BEHAVIOR),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_BEHAVIOR_RV_14), Bytes.toBytes(row.get(2).toString()));
                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_BEHAVIOR),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_BEHAVIOR_BUY_7), Bytes.toBytes(row.get(3).toString()));
                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_BEHAVIOR),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_BEHAVIOR_BUY_14), Bytes.toBytes(row.get(4).toString()));

                        //add data to buffer
                        bufferedMutator.mutate(put);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    //close
                    try {
                        if (bufferedMutator != null) bufferedMutator.close();
                        if (connection != null) connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        JavaRDD<Row> adrdd = adinfo.toJavaRDD();
        adrdd.foreachPartition(new VoidFunction<Iterator<Row>>() {
            @Override
            public void call(Iterator<Row> it) {
                Connection connection = null;
                BufferedMutator bufferedMutator = null;
                try {
                    connection = ConnectionFactory.createConnection(HbaseUtils.getHBaseConfig());
                    bufferedMutator = connection.getBufferedMutator(TableName.valueOf(HbaseUtils.TABLE_NAME_AD));

                    while (it.hasNext()){
                        Row row = it.next();
                        Put put = new Put(Bytes.toBytes(row.get(0).toString()));

                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_AD),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_AD_DIS_1), Bytes.toBytes(row.get(1).toString()));
                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_AD),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_AD_DIS_7), Bytes.toBytes(row.get(2).toString()));
                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_AD),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_AD_CLK_1), Bytes.toBytes(row.get(3).toString()));
                        put.addColumn(Bytes.toBytes(HbaseUtils.FAMILY_AD),
                                Bytes.toBytes(HbaseUtils.QUALIFIER_AD_CLK_7), Bytes.toBytes(row.get(4).toString()));

                        //add data to buffer
                        bufferedMutator.mutate(put);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    //close
                    try {
                        if (bufferedMutator != null) bufferedMutator.close();
                        if (connection != null) connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
    }
}
